#include "data_report.h"

volatile MQTTClient_deliveryToken deliveredtoken;
MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_deliveryToken token;

sqlite3 *db = NULL;
char *errmsg = NULL;

int *device_total;

int main(int argc, char *argv[])
{
    int rc;
    int ch;
    cJSON *item = NULL;
    char *mqtt_addr = NULL;
    int mqtt_port = 0;
    char address[32] = "";

    //权限容错
    if (getuid() != 0)
    {
        execlp("/usr/bin/sudo", "sudo", "./data_report", (char *)NULL);
    }
    //解析点表
    // 打开文件
    FILE *file = NULL;
    file = fopen(FILE_NAME, "r");
    if (NULL == file)
    {
        perror("Open file fail");
        return -1;
    }

    // 获得文件大小
    struct stat statbuf;
    stat(FILE_NAME, &statbuf);
    int fileSize = statbuf.st_size;
    printf("node.json大小：%d\n", fileSize);

    // 分配符合文件大小的内存
    char *jsonStr = (char *)malloc(sizeof(char) * fileSize + 1);
    memset(jsonStr, 0, fileSize + 1);

    // 读取文件中的json字符串
    int size = fread(jsonStr, sizeof(char), fileSize, file);
    if (size == 0)
    {
        printf("读取文件失败！\n");
        fclose(file);
        return -1;
    }

    // 反序列化，获得mqtt服务器的IP和端口
    cJSON *root = cJSON_Parse(jsonStr);
    if (NULL == root)
    {
        printf("Error before: [%s]\n", cJSON_GetErrorPtr());
        free(jsonStr);
        return -1;
    }
    item = cJSON_GetObjectItem(root, "mqtt_server");
    if (NULL != item)
    {
        cJSON *var = NULL;

        var = cJSON_GetObjectItem(item, "addr");
        if (NULL != var && var->type == cJSON_String)
        {
            mqtt_addr = var->valuestring;
        }
        var = cJSON_GetObjectItem(item, "port");
        if (NULL != var && var->type == cJSON_Number)
        {
            mqtt_port = var->valueint;
        }
    }
    //拼接MQTT服务器地址
    sprintf(address, "%s:%d", mqtt_addr, mqtt_port);

    //创建MQTT客户端
    MQTTClient_create(&client, address, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);

    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, root, connlost, msgarrvd, delivered);
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n",
           TOPIC_DOWN, CLIENTID, QOS);
    MQTTClient_subscribe(client, TOPIC_DOWN, QOS);

    //建立数据库
    if (sqlite3_open("./edge_gateway.db", &db) != SQLITE_OK)
    {
        fprintf(stderr, "open err:%s\n", sqlite3_errmsg(db));
        sqlite3_close(db);
        return -1;
    }
    printf("open database success\n");

    //创建表
    const char *sql_create = "CREATE TABLE IF NOT EXISTS sensor_data ("
                             "key INTEGER NOT NULL,"
                             "val TEXT,"
                             "timestamp INTEGER NOT NULL"
                             ");";
    if (sqlite3_exec(db, sql_create, NULL, NULL, &errmsg) != SQLITE_OK)
    {
        fprintf(stderr, "create err:%s\n", errmsg);
        sqlite3_free(errmsg);
        sqlite3_close(db);
        return -1;
    }
    printf("create table ok\n");

    //共享内存
    //初始化
    struct shm_param sp;
    struct std_node *std;
    if (shm_init(&sp, "shm", 256) < 0)
    {
        printf("shm init err\n");
        return -1;
    }
    //映射数据点数量
    device_total = (int *)shm_getaddr(&sp);
    if (NULL == device_total)
    {
        printf("shm getaddr err\n");
        return -1;
    }
    //映射数据点
    std = (struct std_node *)(device_total + 1);

    //解析点表，获得设备信息
    item = cJSON_GetObjectItem(root, "stm32");
    if (NULL != item)
    {
        cJSON *var = NULL;

        var = cJSON_GetObjectItem(item, "data");
        if (NULL != var)
        {
            *device_total = cJSON_GetArraySize(var);

            int i;
            for (i = 0; i < *device_total; i++, std++)
            {
                cJSON *obj = cJSON_GetArrayItem(var, i);
                cJSON *val = NULL;
                if (obj != NULL && obj->type == cJSON_Object)
                {
                    val = cJSON_GetObjectItem(obj, "key");
                    if (val != NULL && val->type == cJSON_Number)
                    {
                        std->key = val->valueint;
                    }

                    val = cJSON_GetObjectItem(obj, "type");
                    if (val != NULL && val->type == cJSON_Number)
                    {
                        std->type = val->valueint;
                    }

                    std->dev_type = 1;
                    switch (std->type)
                    {
                    case 1: //bool
                        std->old_val.b_val = 0;
                        std->new_val.b_val = 0;
                        break;
                    case 2: //int
                        std->old_val.i_val = 0;
                        std->new_val.i_val = 0;
                        break;
                    case 3: //float
                        std->old_val.f_val = 0;
                        std->new_val.f_val = 0;
                        break;
                    }
                    std->ret = -1;
                }
            }
        }
    }

    item = cJSON_GetObjectItem(root, "modbus");
    if (NULL != item)
    {
        cJSON *var = NULL;

        var = cJSON_GetObjectItem(item, "data");
        if (NULL != var)
        {
            int size = cJSON_GetArraySize(var);
            *device_total += size;
            int i;
            for (i = 0; i < size; i++, std++)
            {
                cJSON *obj = cJSON_GetArrayItem(var, i);
                cJSON *val = NULL;
                if (obj != NULL && obj->type == cJSON_Object)
                {
                    val = cJSON_GetObjectItem(obj, "key");
                    if (val != NULL && val->type == cJSON_Number)
                    {
                        std->key = val->valueint;
                    }

                    val = cJSON_GetObjectItem(obj, "type");
                    if (val != NULL && val->type == cJSON_Number)
                    {
                        std->type = val->valueint;
                    }

                    std->dev_type = 2;
                    switch (std->type)
                    {
                    case 1: //bool
                        std->old_val.b_val = 0;
                        std->new_val.b_val = 0;
                        break;
                    case 2: //int
                        std->old_val.i_val = 0;
                        std->new_val.i_val = 0;
                        break;
                    case 3: //float
                        std->old_val.f_val = 0;
                        std->new_val.f_val = 0;
                        break;
                    }
                    std->ret = -1;
                }
            }
        }
    }

    //线程操作
    pthread_t tid_report;
    pthread_t tid_data;
    if (pthread_create(&tid_report, NULL, handler_report, root) != 0)
    {
        perror("pthread report create err");
        return -1;
    }

    if (pthread_create(&tid_data, NULL, handler_data, NULL) != 0)
    {
        perror("pthread data create err");
        return -1;
    }

    do
    {
        ch = getchar();
    } while (ch != 'Q' && ch != 'q');

    //关闭数据库
    if (sqlite3_close(db) != SQLITE_OK)
    {
        fprintf(stderr, "close err:%s\n", sqlite3_errmsg(db));
        return -1;
    }
    //释放资源
    MQTTClient_unsubscribe(client, TOPIC_DOWN);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    cJSON_Delete(root);
    free(jsonStr);
    jsonStr == NULL;
    fclose(file);
    return 0;
}

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

//订阅回调函数、指令处理线程
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    cJSON *cmdroot = NULL;
    cJSON *item = NULL;
    int type = 0;
    struct msgbuf msg;
    struct std_node *std = (struct std_node *)(device_total + 1);
    int report_type = 0;
    int report_period = 0;
    int history_key = 0;
    time_t history_timep[2] = {0}; //time_t == long int

    printf("Message arrived\n");
    printf("topic: %s\n", topicName);

    cmdroot = cJSON_Parse(message->payload);
    if (NULL == cmdroot)
    {
        printf("Error before: [%s]\n", cJSON_GetErrorPtr());
        return -1;
    }
    item = cJSON_GetObjectItem(cmdroot, "type");
    if (NULL != item)
    {
        if (item->type == cJSON_Number)
        {
            type = item->valueint;
        }
    }
    switch (type)
    {
    case 1: //采集指令
        all_report();
        break;
    case 2: //控制指令
        item = cJSON_GetObjectItem(cmdroot, "data");
        if (NULL != item)
        {
            cJSON *var = NULL;

            var = cJSON_GetObjectItem(item, "key");
            if (var != NULL && var->type == cJSON_Number)
            {
                msg.key = var->valueint;
            }

            var = cJSON_GetObjectItem(item, "val");
            if (var != NULL && var->type == cJSON_String)
            {
                strcpy(msg.mdata, var->valuestring);
            }
        }
        for (i = 0; i < *device_total; i++, std++)
        {
            if (msg.key == std->key)
            {
                msg.mtype = std->dev_type;
            }
        }
        if (msg_queue_send("msg", &msg, sizeof(msg), 0) < 0)
        {
            printf("msg_queue_send err\n");
            return -1;
        }
        printf("控制命令发送成功\n");
        break;
    case 3: //模式切换指令
        item = cJSON_GetObjectItem(cmdroot, "data");
        if (NULL != item)
        {
            cJSON *var = NULL;

            var = cJSON_GetObjectItem(item, "type");
            if (var != NULL && var->type == cJSON_Number)
            {
                report_type = var->valueint;
            }

            if (report_type == 2)
            {
                var = cJSON_GetObjectItem(item, "period");
                if (var != NULL && var->type == cJSON_Number)
                {
                    report_period = var->valueint;
                }
            }
        }
        pattern_alter((cJSON *)context, report_type, report_period);
        if (execlp("/usr/bin/sudo", "sudo", "./data_report", (char *)NULL) < 0)
        {
            perror("execlp failed");
            exit(EXIT_FAILURE);
        }
        printf("This line will not be executed if execlp succeeds.\n");
        break;
    case 4: //历史查询指令
        item = cJSON_GetObjectItem(cmdroot, "data");
        if (NULL != item)
        {
            cJSON *var = NULL;

            var = cJSON_GetObjectItem(item, "key");
            if (NULL != var && var->type == cJSON_Number)
            {
                history_key = var->valueint;
            }

            var = cJSON_GetObjectItem(item, "limit");
            if (NULL != var && var->type == cJSON_Array)
            {
                int i;
                int size = cJSON_GetArraySize(var);
                cJSON *arr = NULL;

                for (i = 0; i < size; i++)
                {
                    arr = cJSON_GetArrayItem(var, i);
                    if (arr != NULL && arr->type == cJSON_Number)
                    {
                        history_timep[i] = arr->valueint;
                    }
                }
            }
        }
        char *histr = select_history(history_key, history_timep);
        if (NULL != histr)
        {
            general_report(histr);
            free(histr);
            histr = NULL;
            printf("历史记录查询成功\n");
        }
        break;
    }

    cJSON_Delete(cmdroot);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

//通用推送
void general_report(char *payload)
{
    pubmsg.payload = payload;
    pubmsg.payloadlen = (int)strlen(payload);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client, TOPIC_UP, &pubmsg, &token);
    printf("Finished for publication of %s\n"
           "on topic %s for client with ClientID: %s\n",
           payload, TOPIC_UP, CLIENTID);
    printf("Message with delivery token %d delivered\n", token);
}

//上报线程
void *handler_report(void *arg)
{
    pthread_detach(pthread_self());
    cJSON *root = (cJSON *)arg;
    cJSON *item = NULL;
    int report_type;
    int report_period;
    item = cJSON_GetObjectItem(root, "report");
    if (NULL != item)
    {
        cJSON *var = NULL;

        var = cJSON_GetObjectItem(item, "type");
        if (NULL != var && var->type == cJSON_Number)
        {
            report_type = var->valueint;
        }
        var = cJSON_GetObjectItem(item, "period");
        if (NULL != var && var->type == cJSON_Number)
        {
            report_period = var->valueint;
        }
    }

    if (report_type == 0)
    {
        return NULL;
    }
    else if (report_type == 1) //变化上报
    {
        //先进行一次主动刷新上报，防止有些点不变化导致永远没有机会上报
        all_report();

        int i;
        int flag = 0;
        struct std_node *std = (struct std_node *)(device_total + 1);
        struct std_node *reset_std = std;
        struct timespec ts;
        ts.tv_sec = 0;
        ts.tv_nsec = 500000000;
        char *pubstr = NULL;
        cJSON *pubroot = cJSON_CreateObject();
        cJSON_AddItemToObject(pubroot, "type", cJSON_CreateNumber(1));
        cJSON_AddItemToObject(pubroot, "result", cJSON_CreateNumber(0));

        while (1)
        {
            nanosleep(&ts,NULL);
            cJSON *data = cJSON_CreateArray();
            cJSON_AddItemToObject(pubroot, "data", data);
            std = reset_std;
            for (i = 0; i < *device_total; i++, std++)
            {
                switch (std->type)
                {
                case 1: //bool
                    if (std->new_val.b_val != std->old_val.b_val)
                    {
                        char val[32] = "";
                        std->old_val.b_val = std->new_val.b_val;
                        item = cJSON_CreateObject();
                        cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(std->key));
                        sprintf(val, "%d", std->new_val.b_val);
                        cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
                        cJSON_AddItemToArray(data, item);
                    }
                    break;
                case 2: //int
                    if (std->new_val.i_val != std->old_val.i_val)
                    {
                        char val[32] = "";
                        std->old_val.i_val = std->new_val.i_val;
                        item = cJSON_CreateObject();
                        cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(std->key));
                        sprintf(val, "%d", std->new_val.i_val);
                        cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
                        cJSON_AddItemToArray(data, item);
                    }
                    break;
                case 3: //float
                    if (std->new_val.f_val != std->old_val.f_val)
                    {
                        char val[32] = "";
                        std->old_val.f_val = std->new_val.f_val;
                        item = cJSON_CreateObject();
                        cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(std->key));
                        sprintf(val, "%.1f", std->new_val.f_val);
                        cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
                        cJSON_AddItemToArray(data, item);
                    }
                    break;
                }
            }
            pubstr = cJSON_Print(pubroot);
            int size = cJSON_GetArraySize(data);
            if (size > 0)
            {
                general_report(pubstr);
                free(pubstr);
                pubstr = NULL;
                
                // for (i = 0; i < size; i++)
                // {
                //     cJSON_DeleteItemFromArray(data, i);
                // }
            }
            cJSON_DeleteItemFromObject(pubroot,"data");
        }
        cJSON_Delete(pubroot);
    }
    else if (report_type == 2) //周期上报
    {
        while (1)
        {
            all_report();
            sleep(report_period);
        }
    }
}

//数据存储线程
void *handler_data(void *arg)
{
    pthread_detach(pthread_self());
    while(1)
    {
        //60秒往数据库中存储一次
        insert_data(60);
    }
}

//全部上报
int all_report()
{
    int i;
    cJSON *root = NULL;
    cJSON *item = NULL;
    char *data_str = NULL;

    struct std_node *std = (struct std_node *)(device_total + 1);

    root = cJSON_CreateObject();
    cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
    cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
    cJSON *data = cJSON_CreateArray();

    for (i = 0; i < *device_total; i++, std++)
    {
        item = cJSON_CreateObject();
        cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(std->key));
        if (std->type == 1)
        {
            char val[32] = "";
            sprintf(val, "%d", std->new_val.b_val);
            cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
        }
        else if (std->type == 2)
        {
            char val[32] = "";
            sprintf(val, "%d", std->new_val.i_val);
            cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
        }
        else if (std->type == 3)
        {
            char val[32] = "";
            sprintf(val, "%.1f", std->new_val.f_val);
            cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
        }
        cJSON_AddItemToArray(data, item);
    }
    cJSON_AddItemToObject(root, "data", data);
    data_str = cJSON_Print(root);
    general_report(data_str);
    free(data_str);
    data_str = NULL;
    cJSON_Delete(root);
    return 0;
}

//模式转换
int pattern_alter(cJSON *root, int type, int period)
{
    FILE *file = NULL;
    cJSON *item = NULL;
    item = cJSON_CreateObject();
    cJSON_AddItemToObject(item, "type", cJSON_CreateNumber(type));
    if (type == 2)
    {
        cJSON_AddItemToObject(item, "period", cJSON_CreateNumber(period));
    }
    cJSON_ReplaceItemInObject(root, "report", item);

    file = fopen(FILE_NAME, "w");
    if (NULL == file)
    {
        printf("open file fail\n");
        return -1;
    }
    char *cjValue = cJSON_Print(root);
    int ret = fputs(cjValue, file);
    if (ret == EOF)
    {
        printf("模式修改失败！\n");
        return -1;
    }
    printf("模式修改成功！\n");
    free(cjValue);
    fclose(file);
    return 0;
}

//按照设备号和时间范围查找设备历史值
char *select_history(int history_key, time_t history_timep[])
{
    char *histr = NULL;
    cJSON *hisroot = NULL;
    cJSON *item = NULL;
    int rc = 0;
    sqlite3_stmt *stmt = NULL;
    time_t ts = 0;
    const unsigned char *val = "";

    hisroot = cJSON_CreateObject();
    cJSON_AddItemToObject(hisroot, "type", cJSON_CreateNumber(4));
    cJSON_AddItemToObject(hisroot, "result", cJSON_CreateNumber(0));
    cJSON_AddItemToObject(hisroot, "key", cJSON_CreateNumber(history_key));
    cJSON *data = cJSON_CreateArray();

    //查找数据库
    //对sql语句解析和编译，生成准备语句对象
    const char *sql_select = "SELECT * FROM sensor_data WHERE key = ? AND timestamp > ? AND timestamp < ?";
    rc = sqlite3_prepare_v2(db, sql_select, strlen(sql_select), &stmt, NULL);
    if (rc != SQLITE_OK)
    {
        fprintf(stderr, "Prepare select statement error: %s\n", sqlite3_errmsg(db));
        sqlite3_close(db);
        return NULL;
    }

    // 绑定参数
    rc = sqlite3_bind_int(stmt, 1, history_key);
    if (rc != SQLITE_OK)
    {
        fprintf(stderr, "Bind history_key error: %s\n", sqlite3_errmsg(db));
        sqlite3_finalize(stmt);
        sqlite3_close(db);
        return NULL;
    }
    rc = sqlite3_bind_int64(stmt, 2, history_timep[0]);
    if (rc != SQLITE_OK)
    {
        fprintf(stderr, "Bind history_timep[0] error: %s\n", sqlite3_errmsg(db));
        sqlite3_finalize(stmt);
        sqlite3_close(db);
        return NULL;
    }
    rc = sqlite3_bind_int64(stmt, 3, history_timep[1]);
    if (rc != SQLITE_OK)
    {
        fprintf(stderr, "Bind history_timep[1] error: %s\n", sqlite3_errmsg(db));
        sqlite3_finalize(stmt);
        sqlite3_close(db);
        return NULL;
    }

    // 执行查询
    while (sqlite3_step(stmt) == SQLITE_ROW)
    {
        ts = sqlite3_column_int64(stmt, 2);
        val = sqlite3_column_text(stmt, 1);
        item = cJSON_CreateObject();
        cJSON_AddItemToObject(item, "time", cJSON_CreateNumber(ts));
        cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
        cJSON_AddItemToArray(data, item);
    }

    // 清理
    sqlite3_finalize(stmt);

    cJSON_AddItemToObject(hisroot, "data", data);
    histr = cJSON_Print(hisroot);
    cJSON_Delete(hisroot);

    return histr;
}

//周期存储到数据库
int insert_data(int sleep_sec)
{
    struct std_node *std = (struct std_node *)(device_total + 1);
    int i;
    sqlite3_stmt *stmt = NULL;
    int rc;
    struct database_t dbt[9];
    time_t timep;
    for (i = 0; i < *device_total; i++, std++)
    {
        dbt[i].key = std->key;
        switch (std->type)
        {
        case 1:
            sprintf(dbt[i].val, "%d", std->new_val.b_val);
            break;
        case 2:
            sprintf(dbt[i].val, "%d", std->new_val.i_val);
            break;
        case 3:
            sprintf(dbt[i].val, "%.1f", std->new_val.f_val);
            break;
        }
        time(&timep);
        dbt[i].timestamp = timep;
    }

    const char *sql_insert = "INSERT INTO sensor_data  VALUES(?,?,?);";

    rc = sqlite3_prepare_v2(db, sql_insert, strlen(sql_insert), &stmt, NULL);
    if (rc != SQLITE_OK)
    {
        fprintf(stderr, "Prepare insert statement error: %s\n", sqlite3_errmsg(db));
        sqlite3_close(db);
        return -1;
    }

    for (i = 0; i < *device_total; i++)
    {
        rc = sqlite3_bind_int(stmt, 1, dbt[i].key);
        if (rc != SQLITE_OK)
        {
            fprintf(stderr, "Bind dbt[%d].key error: %s\n", i, sqlite3_errmsg(db));
            sqlite3_finalize(stmt);
            sqlite3_close(db);
            return -1;
        }

        rc = sqlite3_bind_text(stmt, 2, dbt[i].val,strlen(dbt[i].val),NULL);
        if (rc != SQLITE_OK)
        {
            fprintf(stderr, "Bind dbt[%d].val error: %s\n", i, sqlite3_errmsg(db));
            sqlite3_finalize(stmt);
            sqlite3_close(db);
            return -1;
        }

        rc = sqlite3_bind_int64(stmt, 3, dbt[i].timestamp);
        if (rc != SQLITE_OK)
        {
            fprintf(stderr, "Bind dbt[%d].timestamp error: %s\n", i, sqlite3_errmsg(db));
            sqlite3_finalize(stmt);
            sqlite3_close(db);
            return -1;
        }
        sqlite3_step(stmt);
        sqlite3_reset(stmt);
    }
    sqlite3_finalize(stmt);
    printf("%d个设备的数据存储成功\n",*device_total);

    sleep(sleep_sec);
}